package com.wsx.ones.cloud.kafka;

import com.google.gson.Gson;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created by wangshuaixin on 17/6/21.
 */
public class KafkaProdu {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConstant.class);

    private static final String PRODUCER_TYPE = "async";

    private static Producer<String, String> producer;

    static {
        getProducerConfig();
    }

    public static void getProducerConfig() {
        Properties props = new Properties();


        //props.put(KafkaConstant.BROKERLIST, KafkaConfigUtils.getProps(KafkaConstant.BROKERLIST));

        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test_group");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        props.put(KafkaConstant.REQUESTACKS, "1");
        props.put(KafkaConstant.REQUESTTIMEOUTMS, "10000");
        props.put(KafkaConstant.CODEC, "none");

        // 消息发送类型，sync同步，async异步
        /**
        String producerType = KafkaConfigUtils.getProps(KafkaConstant.PRODUCERTYPE);
        if (StringUtils.equalsIgnoreCase(producerType, PRODUCER_TYPE)) {
            props.put(KafkaConstant.PRODUCERTYPE, producerType);
            props.put(KafkaConstant.QUEUEMAXMS,
                    KafkaConfigUtils.getProps(KafkaConstant.QUEUEMAXMS, "5000"));
            props.put(KafkaConstant.QUEUEMAXMESSAGE,
                    KafkaConfigUtils.getProps(KafkaConstant.QUEUEMAXMESSAGE, "10000"));
            props.put(KafkaConstant.QUEUETIMEOUTMS,
                    KafkaConfigUtils.getProps(KafkaConstant.QUEUETIMEOUTMS, "-1"));
            props.put(KafkaConstant.QUEUEBATCHNUM,
                    KafkaConfigUtils.getProps(KafkaConstant.QUEUEBATCHNUM, "200"));
        }
         **/

        producer = new KafkaProducer<String, String>(props);
    }

    /**
     *
     * 功能描述: <br>
     * 使用kafkatopic进行发送
     *
     * @param obj
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    public static void sendMessage(Object obj) {
        String jsonMsg = new Gson().toJson(obj);
        LOGGER.info("try to send data to bigdata:{}", jsonMsg);
        ProducerRecord<String, String> data = new ProducerRecord<String, String>(KafkaConstant.TOPIC, jsonMsg);
        producer.send(data);
    }

    /**
     *
     * 功能描述: <br>
     * 使用指定的kafkatopic进行发送
     *
     * @param jsonMsg
     * @param kafkaTopic
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    public static void sendMessage(String jsonMsg, String kafkaTopic) {
        LOGGER.debug("try to send member sensitive data to bigdata:{}", jsonMsg);
        ProducerRecord<String, String> data = new ProducerRecord<String, String>(kafkaTopic, jsonMsg);
        producer.send(data);
    }

    public static void main(String[] args) {
        for (int i = 0; i < 1000; i++) {
            String value = "hello :" + i;
            KafkaProdu.sendMessage(value, "test");
        }
    }
}
